package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

/**
 * @author Felix
 * @date 2024/7/26
 * 对关联后的数据进行处理
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {
    private Map<String,TableProcessDim> configMap = new HashMap<>();
    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置信息预加载到程序中
        Connection mysqlConnection = JdbcUtil.getMysqlConnection();
        String sql = "select * from gmall0221_config.table_process_dim";
        List<TableProcessDim> tableProcessDimList = JdbcUtil.queryList(mysqlConnection, sql, TableProcessDim.class, true);
        for (TableProcessDim tableProcessDim : tableProcessDimList) {
            configMap.put(tableProcessDim.getSourceTable(),tableProcessDim);
        }
        JdbcUtil.closeConnection(mysqlConnection);
    }

    //处理主流业务数据    根据表名到广播状态中获取对应的配置，如果获取到了，说明是维度，将维度数据传递到下游
    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
        //获取处理的业务数据库表的表名
        String table = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //判断当前处理的表名在不在广播状态中
        TableProcessDim tableProcessDim = null;


        if((tableProcessDim = broadcastState.get(table)) != null
                ||(tableProcessDim = configMap.get(table)) != null){
            //说明处理的数据是维度数据，取出其中data部分内容，向下游传递
            //{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"abcbbb","id":1}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //在向下游传递数据前，过滤掉不需要传递的字段
            //{"tm_name":"Redmi","id":1}
            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteNotNeedColumn(dataJsonObj,sinkColumns);

            //在向下游传递数据前，补充type字段
            //{"tm_name":"Redmi","id":1,"type":"update"}
            String type = jsonObj.getString("type");
            dataJsonObj.put("type", type);

            out.collect(Tuple2.of(dataJsonObj,tableProcessDim));
        }
    }

    //处理广播流配置数据  将配置信息放到广播状态中或者从广播状态中删除对应的配置信息  K:维度表名  V:实体类对象
    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context ctx, Collector<Tuple2<JSONObject, TableProcessDim>> out) throws Exception {
        //获取对配置表进行的操作的类型
        String op = tableProcessDim.getOp();
        //获取广播状态
        BroadcastState<String, TableProcessDim> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //获取维度表名称
        String sourceTable = tableProcessDim.getSourceTable();

        if("d".equals(op)){
            //说明从配置表中删除了一条配置，将对应的配置信息从广播状态中也删除掉
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else{
            //说明对配置表做了添加、读取、更新操作，将最新的配置信息放到广播状态中
            broadcastState.put(sourceTable,tableProcessDim);
            configMap.put(sourceTable,tableProcessDim);
        }
    }

    //过滤掉不需要传递的字段
    //dataJsonObj: {"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"abcbbb","id":1}
    //sinkColumnS: id,tm_name
    private static void deleteNotNeedColumn(JSONObject dataJsonObj, String sinkColumns) {
        /*
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        JSONObject newJsonObj = new JSONObject();
        for (String column : columnList) {
            newJsonObj.put(column,dataJsonObj.getString(column));
        }
        return newJsonObj;
        */
        /*String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        for (Map.Entry<String, Object> entry : entrySet) {
            if(!columnList.contains(entry.getKey())){
                entrySet.remove(entry);
            }
        }

        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        Iterator<Map.Entry<String, Object>> it = entrySet.iterator();
        for (;it.hasNext();){
            Map.Entry<String, Object> entry = it.next();
            if(!columnList.contains(entry.getKey())){
                it.remove();
            }
        }
         */
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();

        entrySet.removeIf(entry-> !columnList.contains(entry.getKey()));
    }

}
